Module 6: Event-Driven Development Using Queues and Streams

Module Overview

This module covers how Lambda processes events from queues (SQS) and streams (Kinesis, DynamoDB Streams) using event source mappings (polling-based invocation).

Topics

• Event source mappings (polling pattern)
• Amazon SQS as a Lambda event source
• Kinesis Data Streams as a Lambda event source
• DynamoDB Streams as a Lambda event source
• Queues vs Streams - when to use each
• Error handling, batching, and scaling behavior

1. Event Source Mappings (Polling Pattern)

Unlike async invocations (EventBridge/SNS push to Lambda), queues and streams require Lambda to poll for records.

How It Works

Key Points

• Lambda polls the source (you don't configure triggers on the queue itself)
• Lambda invokes your function synchronously with a batch of records
• The event source mapping handles polling, batching, and retry logic
• You configure: batch size, batch window, concurrency, error handling

2. Amazon SQS as Lambda Event Source

Queues provide a buffer between a producer and a single consumer. Messages are processed and deleted.

How Lambda Processes SQS

SQS Queue Types

SAM Template: SQS + Lambda

SQS Processing Lifecycle

SAM Template: SQS + Lambda

Resources:
  OrderQueue:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeout: 180   # 6x Lambda timeout (30s)
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt OrderDLQ.Arn
        maxReceiveCount: 3

  OrderDLQ:
    Type: AWS::SQS::Queue

  ProcessOrderFn:
    Type: AWS::Serverless::Function
    Properties:
      Handler: app.handler
      Runtime: python3.12
      Timeout: 30
      Events:
        SQSEvent:
          Type: SQS
          Properties:
            Queue: !GetAtt OrderQueue.Arn
            BatchSize: 10
            FunctionResponseTypes:
              - ReportBatchItemFailures

3. Kinesis Data Streams & DynamoDB Streams

Streams buffer data arriving continuously and support multiple consumers. Data is retained (not deleted on read).

How Lambda Processes Streams

Kinesis vs DynamoDB Streams

Scaling Behavior Comparison

Kinesis Stream Processing - Shard-Based

Error Handling for Streams

# Event source mapping configuration for streams
EventSourceMapping:
  Type: AWS::Lambda::EventSourceMapping
  Properties:
    EventSourceArn: !GetAtt MyStream.Arn
    FunctionName: !Ref ProcessStreamFn
    StartingPosition: LATEST
    BatchSize: 100
    MaximumBatchingWindowInSeconds: 5
    ParallelizationFactor: 2          # 2 concurrent per shard
    MaximumRetryAttempts: 3           # Don't block forever
    BisectBatchOnFunctionError: true  # Split batch to find bad record
    MaximumRecordAgeInSeconds: 86400  # Skip records older than 1 day
    DestinationConfig:
      OnFailure:
        Destination: !GetAtt FailedRecordsSQS.Arn  # Send failures to DLQ

4. Queues vs Streams - When to Use Each

5. Live Demo: SQS + Lambda with DLQ

Step 1: Create SQS queue with DLQ

# Create dead-letter queue
aws sqs create-queue --queue-name demo-orders-dlq

# Get DLQ ARN
DLQ_ARN=$(aws sqs get-queue-attributes --queue-url https://sqs.us-west-2.amazonaws.com/ACCOUNT/demo-orders-dlq \
  --attribute-names QueueArn --query 'Attributes.QueueArn' --output text)

# Create main queue with redrive policy
aws sqs create-queue --queue-name demo-orders \
  --attributes '{
    "VisibilityTimeout": "180",
    "RedrivePolicy": "{\"deadLetterTargetArn\":\"'$DLQ_ARN'\",\"maxReceiveCount\":\"3\"}"
  }'

Step 2: Create Lambda function

# handler.py
import json

def handler(event, context):
    failed_ids = []
    for record in event["Records"]:
        body = json.loads(record["body"])
        order_id = body.get("orderId", "unknown")
        print(f"Processing order: {order_id}")
        
        # Simulate failure for certain orders
        if body.get("fail"):
            failed_ids.append(record["messageId"])
            continue
        
        print(f"  Success: {order_id}")
    
    # Report partial batch failures
    return {
        "batchItemFailures": [
            {"itemIdentifier": mid} for mid in failed_ids
        ]
    }

Step 3: Create event source mapping

aws lambda create-event-source-mapping \
  --function-name demo-order-processor \
  --event-source-arn arn:aws:sqs:us-west-2:ACCOUNT:demo-orders \
  --batch-size 5 \
  --function-response-types ReportBatchItemFailures

Step 4: Send test messages

# Success message
aws sqs send-message --queue-url https://sqs.us-west-2.amazonaws.com/ACCOUNT/demo-orders \
  --message-body '{"orderId":"ORD-001","amount":99.99}'

# Failure message (will retry then go to DLQ)
aws sqs send-message --queue-url https://sqs.us-west-2.amazonaws.com/ACCOUNT/demo-orders \
  --message-body '{"orderId":"ORD-002","amount":50,"fail":true}'

Step 5: Check DLQ after failures

aws sqs receive-message --queue-url https://sqs.us-west-2.amazonaws.com/ACCOUNT/demo-orders-dlq

6. What's New (2024-2025 Updates)

• Partial Batch Response (SQS) - Report individual failed messages instead of retrying the entire batch
• Stream Checkpointing - Resume from a specific sequence number on failure (skip already-processed records)
• Maximum Concurrency for SQS - Cap Lambda concurrency per event source mapping (prevent downstream overload)
• Parallelization Factor (Streams) - Process up to 10 batches per shard simultaneously for higher throughput
• Tumbling Windows (Streams) - Aggregate records over time windows (1s-15min) before invoking Lambda
• Event Filtering for SQS - Filter messages before invoking Lambda (reduce invocations and cost)

7. Module Summary